package com.mavenq.fly.business.service;

import com.mavenq.fly.business.entity.TempData;
import com.mavenq.fly.business.mapper.TempDataMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Service
public class DataConsumerService {

    private final AtomicBoolean running = new AtomicBoolean(false);

    @Value("${app.consumer.poll-timeout:1}")
    private long POLL_TIMEOUT_SECONDS;

    @Value("${app.consumer.max-retry:3}")
    private int MAX_RETRY;

    @Autowired
    private TempDataMapper tempDataMapper;

    @Autowired
    private BlockingQueue<TempData> dataQueue;

    private Thread consumerThread;

    public synchronized void start() {
        if (running.get()) {
            log.warn("Consumer is already running");
            return;
        }

        running.set(true);
        consumerThread = new Thread(this::consumeData, "Data-Consumer");
        consumerThread.start();
        log.info("Data consumer started");
    }

    public synchronized void stop() {
        running.set(false);
        if (consumerThread != null) {
            consumerThread.interrupt();
        }
        log.info("Data consumer stopped");
    }

    private void consumeData() {
        while (running.get() || !dataQueue.isEmpty()) {
            try {
                TempData data = dataQueue.poll(POLL_TIMEOUT_SECONDS, TimeUnit.SECONDS);
                if (data == null) continue;

                processData(data);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.info("Consumer thread interrupted");
                break;
            } catch (Exception e) {
                log.error("Error in consumer loop: {}", e.getMessage(), e);
            }
        }
        log.info("Data consumer stopped, queue size: {}", dataQueue.size());
    }

    @Transactional
    public void processData(TempData data) {
        try {
            // 1. 执行业务逻辑
            doBusinessProcessing(data);

            // 2. 归档数据
            tempDataMapper.archiveData(data.getId());

            // 3. 删除原数据
            tempDataMapper.deleteData(data.getId());

            log.debug("Processed data successfully: {}", data.getId());
        } catch (Exception e) {
            handleProcessingFailure(data, e);
            throw e; // 触发事务回滚
        }
    }

    private void doBusinessProcessing(TempData data) {
        // 实现具体的业务逻辑
        log.info("Processing data: {}", data.getId());
        // 模拟业务处理
        try {
            TimeUnit.MILLISECONDS.sleep(200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Processing interrupted", e);
        }
    }

    private void handleProcessingFailure(TempData data, Exception e) {
        String errorMsg = e.getMessage();
        if (errorMsg != null && errorMsg.length() > 500) {
            errorMsg = errorMsg.substring(0, 500);
        }

        if (data.getRetryCount() >= MAX_RETRY) {
            tempDataMapper.markAsFailed(
                    data.getId(),
                    TempData.Status.FAILED.name(),
                    "Max retry exceeded: " + errorMsg
            );
            log.error("Data processing failed after max retries: {}", data.getId(), e);
        } else {
            tempDataMapper.markAsFailed(
                    data.getId(),
                    TempData.Status.PENDING.name(),
                    "Retryable error: " + errorMsg
            );
            log.warn("Data processing failed, will retry: {}", data.getId(), e);
        }
    }
}